package main

import (
	"context"
	"fmt"
	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/clientv3/concurrency"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"google.golang.org/grpc"
	"log"
)

import (
	"time"
)

func initEtcd() {
	// etcdctl put /tasks/task1 '{"id": 1, "spec": "*/5 * * * * *", "command": "php -v"}'
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"http://39.100.58.10:2379"},
		DialTimeout: 3 * time.Second,
		DialOptions: []grpc.DialOption{grpc.WithBlock()},
	})

	if err != nil {
		fmt.Println(err)
		return
	}

	lock("lock", client)

	kv := clientv3.NewKV(client)
	getResponse, err := kv.Get(context.TODO(), "/tasks/", clientv3.WithPrefix())
	if err != nil {
		fmt.Println(err)
		return
	}

	for _, item := range getResponse.Kvs {
		fmt.Println(item)
	}

	startRevision := getResponse.Header.Revision + 1
	go watch(startRevision, client)
}

func lock(lockKey string, client *clientv3.Client) {
	session, err := concurrency.NewSession(client)
	if err != nil {
		log.Fatal(err)
	}
	mutex := concurrency.NewMutex(session, lockKey)
	if err := mutex.Lock(context.TODO()); err != nil {
		log.Fatal("go get mutex failed " + err.Error())
	}
	fmt.Println("go get mutex sucess")
	fmt.Println(mutex)
	fmt.Println(mutex.IsOwner().Result)

	time.Sleep(time.Duration(10) * time.Second)
	mutex.Unlock(context.TODO())
	fmt.Println("go release lock")
}

func watch(startRevision int64, client *clientv3.Client) {
	watcher := clientv3.NewWatcher(client)

	//var watchResponseChan <-chan clientv3.WatchResponse
	watchResponseChan := watcher.Watch(context.TODO(), "/cron/jobs/", clientv3.WithPrefix(), clientv3.WithRev(startRevision))

	for res := range watchResponseChan {
		for _, event := range res.Events {
			switch event.Type {
			case mvccpb.PUT:
				fmt.Println("修改为：", string(event.Kv.Value), "Revision:", event.Kv.CreateRevision, event.Kv.ModRevision)
			case mvccpb.DELETE:
				fmt.Println("删除了：", "Revision:", event.Kv.ModRevision)
			}
		}
	}
}
